package com.hbwxz.rabbitmq;

import com.hbwxz.constant.Constants;
import com.hbwxz.utils.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 面试题三：订单不支付，15分钟订单失效，用 RabbitMQ 怎么设计
 * <p>
 * 修改了交换机 or 队列的参数时需要每次删除 交换机 和 队列，不然会报错
 * <p>
 * 想法 or 思路：用 RabbitMQ 来实现一个延时队列的效果，RabbitMQ 中可以给队列 queue 设置过期时间，这样设置的是整个队列当中所有消息的一个过期时间，时间一到就会进入死信队列，再加一个消费者来监听死信队列
 * PS：延时队列是没有消费者的，只有过期时间，消息也可以单独设计过期时间，如果都设置了，取值 = min(队列过期时间, 消息过期时间)；
 * 链路：延时队列 -> 消息过期 -> 死信队列 -> 消费者 -> 校验订单是否支付，支付下一条消息，没有支付走取消订单的一个业务逻辑
 *
 * @author shenzw
 * @date 2023/11/9
 */
public class RabbitMqFaceThree {
    public static void main(String[] args) throws IOException, TimeoutException {

        // 获取 channel
        Channel channel = RabbitMqUtils.getChannel();

        /// ======================================================= 创建死信队列（全称：Dead-Letter-Exchange） ================================================================= ///
        // PS：如果交换机设置的是 fanout 类型，就不需要指定路由 Key，如果用的是 Direct 类型，就需要指定路由 Key；x-dead-letter-routing-key
        channel.exchangeDeclare(Constants.DEAD_EXCHANGE, Constants.DIRECT_EXCHANGE, true, false, null); // 交换机的名字，交换机的类型，是否持久化，不知道，参数
        channel.queueDeclare(Constants.DEAD_QUEUE, true, false, false, null);
        channel.queueBind(Constants.DEAD_QUEUE, Constants.DEAD_EXCHANGE, Constants.DEAD_ROUTING_KEY); // 队列，交换机，routingKey

        /// ======================================================= 创建延时队列（全称：Dead-Letter-Exchange） ================================================================= ///
        // 小 tips：fanout 类型交换机不需要指定 routingKey，因为 fanout 类型的交换机会将消息发送与所有与之相关的队列上
        channel.exchangeDeclare(Constants.MY_EXCHANGE, Constants.DIRECT_EXCHANGE, true); // 交换机的名字，交换机的类型，是否持久化
        Map<String, Object> argsDelay = new HashMap<>(10);
        argsDelay.put("x-message-ttl", 10 * 1000);
        argsDelay.put("x-dead-letter-exchange", Constants.DEAD_EXCHANGE);
        argsDelay.put("x-dead-letter-routing-key", Constants.DEAD_ROUTING_KEY);
        channel.queueDeclare(Constants.MY_QUEUE, true, false, false, argsDelay);
        channel.queueBind(Constants.MY_QUEUE, Constants.MY_EXCHANGE, Constants.MY_ROUTING_KEY); // 队列，交换机，routingKey

        // 我的生产者
        myProducer(channel);

        // 我的消费者
        myConsumer(channel);

//        RabbitMqUtils.close(channel);

    }

    /**
     * 我的消费者
     *
     * @param channel 信道
     */
    private static void myConsumer(Channel channel) throws IOException {
        // 消费消息（队列，是否自动确认消费消息【关闭，不然消息没有正常消费的情况下也会自动确认】，交换机，消费者【具体执行消费的相关逻辑】）
        channel.basicConsume(Constants.DEAD_QUEUE, false, Constants.DEAD_EXCHANGE, new DefaultConsumer(channel) {
            /**
             * 消费消息的核心方法，覆盖父类方法
             * @param consumerTag 消息者 ID
             * @param envelope Envelope（存放消息唯一 ID 等一些相关信息）
             * @param properties AMQP.BasicProperties
             * @param body 消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                long deliveryTag = envelope.getDeliveryTag(); // 唯一的 ID
                System.out.println("Dead-letter-exchange handleDelivery receive message = " + new String(body) + "，deliveryTag = " + deliveryTag);
                channel.basicAck(deliveryTag, true); // 手动 Ack
            }
        });
    }

    /**
     * 我的生产者
     *
     * @param channel 信道
     */
    private static void myProducer(Channel channel) throws IOException {

        byte[] msg = "Hello RabbitMQ".getBytes();

        // 交换机名称，路由 KEY，mandatory， AMQP.BasicProperties()，发送消息
        // 模拟演示“我的交换机”失败路由到“我的队列”，然后消息路由到备份交换机中
        channel.basicPublish(Constants.MY_EXCHANGE, Constants.MY_ROUTING_KEY, false,
                new AMQP.BasicProperties().builder()
                        .deliveryMode(2) // 消息持久化设置
                        .contentType("text/plain")
                        .build(),
                msg);
    }
}
